#include "Connector.h"
#include "Socket.h"
#include "sock.h"
#include "eventloop/Eventloop.h"
#include "eventloop/IOchannel.h"
#include "log/Logger.h"
#include "utility/assertion.h"
#include "utility/feature.h"
#include "wrap/error.h"
#include "wrap/unistd.h"

namespace _ {

constexpr double INIT_RETRY_DELAY = 1.0;
constexpr double MAX_RETRY_DELAY = 30.0;

}

Connector:: Connector(Eventloop &loop, Sockaddr addr, bool retry):
    std::enable_shared_from_this<Connector>{},
    _newConnectionCallback{[](std::unique_ptr<Socket> && /*socket*/){}},
    _loop{loop},
    _channel{},
    _retryDelay{_::INIT_RETRY_DELAY},
    _addr{addr},
    _state{_State::_DISCONNECTED},
    _connecting{false},
    _retry{retry} {
    LOG_DEBUG << "constructed " << _addr.IP_port() << ' ' << this;
}

Connector::~Connector() {
    LOG_DEBUG << "destructed " << _addr.IP_port() << ' ' << this;
    assert(!_channel);
}

void Connector::set_new_connection_callback(const NEWConnectionCallback &callback) {
    assert(callback);
    _newConnectionCallback = callback;
}

    
void Connector::restart() {
    _loop.ASSERT_IN_THIS_THREAD();
    _state = _State::_DISCONNECTED;
    _retryDelay = _::INIT_RETRY_DELAY;
    _connecting = true;
    _start_in_loop();
}

void Connector::start() {
    _connecting = true;
    _loop.run_in_loop([this]() { _start_in_loop(); });
}

void Connector::stop() {
    _connecting = false;
    _loop.queue_in_loop([this](){
        _loop.ASSERT_IN_THIS_THREAD();
        if (_state != _State::_CONNECTING) {
            return;
        }
        _state = _State::_DISCONNECTED;
        _reconnect(_remove_and_reset_channel());
    });
}


void Connector::_after_connect(int fd, int connectResult) {
    switch (connectResult) {
    case EINPROGRESS:
        ::printf("connection can't be completed immediately, may try again.\n");
        FALLTHROUGH();
    case 0:
    case EINTR:
    case EISCONN:
        _state = _State::_CONNECTING;
        assert(!_channel);
        _channel = std::make_unique<IOchannel>(
            _loop, fd, "Connector-" + _addr.IP_port());
        _channel->set_write_callback([this](){ _write_callback(); });
        _channel->set_error_callback([this](){ _error_callback(); });
        _channel->enable_writing();
        break;
    case EAGAIN:
    case EADDRINUSE:
    case EADDRNOTAVAIL:
    case ECONNREFUSED:
    case ENETUNREACH:
        _reconnect(fd);
        break;
    case EACCES:
    case EPERM:
    case EAFNOSUPPORT:
    case EALREADY:
    case EBADF:
    case EFAULT:
    case ENOTSOCK:
        LOG_SYSERROR << "connector error: " << connectResult;
        Wrap::close_(fd);
        break;
    case EINVAL:
        LOG_WARN << "EINVAL";
        _reconnect(fd);
        break;
    default:
        LOG_SYSERROR << "Unexpected error: " << connectResult;
        Wrap::close_(fd);
        break;
    }
}

void Connector::_after_connected(int fd) {
    _state = _State::_CONNECTED;
    if (!_connecting) {
        Wrap::close_(fd);
        return;
    }
    _newConnectionCallback(std::make_unique<Socket>(fd));
}

void Connector::_error_callback() {
    LOG_ERROR << "state=" << _state_to_string();
    if (_state != _State::_CONNECTING) {
        return;
    }
    const auto fd = _remove_and_reset_channel();
    const auto err = Sock::socket_error(fd);
    LOG_VERBOSE << "SO_ERROR=" << err << ' ' << Wrap::strerror_r_(err);
    _reconnect(fd); 

}

int Connector::_remove_and_reset_channel() {
    _channel->disable();
    _channel->remove();
    const auto fd = _channel->fd();
    _loop.queue_in_loop([this](){ _channel.reset(); });
    return fd;
}

void Connector::_reconnect(int fd) {
    Wrap::close_(fd);
    _state = _State::_DISCONNECTED;
    if (!_connecting || _retryDelay >= _::MAX_RETRY_DELAY || !_retry) {
        LOG_WARN << "stop connecting to " << _addr.IP_port();
        return;
    }
    LOG_INFO << "Connector retry connecting to " << _addr.IP_port() << " in "
             << _retryDelay << 's';
    _loop.run_after(_retryDelay, [sharedThis = shared_from_this()](){
        sharedThis->_start_in_loop();
    });
    _retryDelay *= 2;
}

void Connector::_start_in_loop() {
    _loop.ASSERT_IN_THIS_THREAD();
    assert(_state == _State::_DISCONNECTED);
    if (!_connecting) {
        LOG_DEBUG << "do not connect";
        return;
    }
    const auto fd = Sock::create_nonblock_socket(_addr.family());
    _after_connect(fd, Sock::connect(fd, _addr.sock_addr()) != 0 ? errno : 0);
}

void Connector::_write_callback() {
    LOG_VERBOSE << "state=" << _state_to_string();
    if (_state != _State::_CONNECTING) {
        assert(_state == _State::_DISCONNECTED);
        return;
    }
    const auto fd = _remove_and_reset_channel();
    const auto err = Sock::socket_error(fd);
    if (err != 0) {
        LOG_WARN << "SO_ERROR=" << err << ' ' << Wrap::strerror_r_(err);
        _reconnect(fd);
        return;
    }
    if (Sock::is_connect_to_self(fd)) {
        LOG_WARN << "connect to self";
        _reconnect(fd);
        return;
    }
    _after_connected(fd);
}

const char *Connector::_state_to_string() const {
    switch (_state) {
    case _State::_DISCONNECTED:
        return "_DISCONNECTED";
    case _State::_CONNECTING:
        return "_CONNECTING";
    case _State::_CONNECTED:
        return "_CONNECTED";
    default:
        return "UNKNOWN STATE";
    }
}